package com.bs.service;

import com.bs.domain.ServerExecuteInfo;
import com.bs.service.callback.ModuleNameCallBack;
import com.bs.utils.DataRowCollection;
import com.bs.utils.DataTable;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.InternalTerms;
import org.elasticsearch.search.aggregations.bucket.terms.LongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.aggregations.metrics.stats.extended.ExtendedStatsBuilder;
import org.elasticsearch.search.aggregations.metrics.stats.extended.InternalExtendedStats;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Service
public class ServerAnalysisService {

    @Autowired
    ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    private ModuleNameService moduleNameService;

    private StringBuilder svcNames = new StringBuilder();
    private String partition;
    private String startTime;
    private String endTime;

    private Map<String, String> svcSerMap = new HashMap<>();

    private SearchResponse getResponse() {
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "4. service end."));
        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("partition", partition))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch("logstash-*-svcsql-*").setTypes("log");
        TermsBuilder serverAgg = AggregationBuilders.terms("servers").field("pid")
            .size(30)
            .order(Terms.Order.aggregation("svr_exec_time", "std_deviation", false));
        ExtendedStatsBuilder extAgg = AggregationBuilders.extendedStats("svr_exec_time").field("exec_time");
        TermsBuilder serviceAgg = AggregationBuilders.terms("services").field("svc");
        TermsBuilder hostAgg = AggregationBuilders.terms("hosts").field("host.raw");
        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .addAggregation(serverAgg.subAggregation(extAgg).subAggregation(serviceAgg).subAggregation(hostAgg))
            .execute().actionGet();

    }

    public List<ServerExecuteInfo> getServerList(String partition, String startTime, String endTime, String svcName) {
        this.partition = partition;
        this.startTime = startTime;
        this.endTime = endTime;
        Map<ServerExecuteInfo, String> serverExecuteInfoMap = new HashMap<>();
        SearchResponse response = getResponse();
        Map<String, Aggregation> aggMap = response.getAggregations().asMap();
        LongTerms serverAgg = (LongTerms) aggMap.get("servers");
        List<Terms.Bucket> serverBucket = serverAgg.getBuckets();
        serverBucket.stream().forEach(buck->{
            long pid = (Long) buck.getKey();
            long count = buck.getDocCount();
            Map<String, Aggregation> internalAgg = buck.getAggregations().asMap();
            InternalExtendedStats internalExtendedStats = (InternalExtendedStats) internalAgg.get("svr_exec_time");
            InternalTerms internalTerms = (InternalTerms) internalAgg.get("services");
            InternalTerms internalTerms1 = (InternalTerms) internalAgg.get("hosts");
            double min = internalExtendedStats.getMin();
            double max = internalExtendedStats.getMax();
            double avg = internalExtendedStats.getAvg();
            double std_deviation = internalExtendedStats.getStdDeviation();
            List<Terms.Bucket> svcList = internalTerms.getBuckets();
            List<Terms.Bucket> hostBuck = (List<Terms.Bucket>) internalTerms1.getBuckets();
            String hostName = hostBuck.size()>0?hostBuck.get(0).getKeyAsString():" ";
            ServerExecuteInfo serverExecuteInfo = new ServerExecuteInfo(pid, null, count, min, max, avg, std_deviation, hostName);
            setServerExecuteInfoMap(svcList, svcName, serverExecuteInfo, serverExecuteInfoMap);
        });


        return setServerName(serverExecuteInfoMap);
    }

    private void setServerExecuteInfoMap(List<Terms.Bucket> svcList, String svcName, ServerExecuteInfo serverExecuteInfo, Map<ServerExecuteInfo, String> serverExecuteInfoMap) {
        String finalSvcName = svcName;
        boolean isExist = svcList.stream().anyMatch(bucket -> bucket.getKeyAsString().equals(finalSvcName));
        String svc_name;
        if (svcName==null)svcName="";
        if (!"".equals(svcName) && !isExist) {
            return;
        } else if ("".equals(svcName)) {
            svc_name = svcList.get(0).getKeyAsString();
            svcNames.append("'").append(svc_name).append("'").append(",");
        } else {
            svc_name = svcName;
            svcNames.append(svc_name);
        }
        serverExecuteInfoMap.put(serverExecuteInfo, svc_name);

    }


    private List<ServerExecuteInfo> setServerName(Map<ServerExecuteInfo, String> serverExecuteInfoMap) {
        getSvcSvrMap();
        List<ServerExecuteInfo> serverExecuteInfoList = new ArrayList<>();
        for (ServerExecuteInfo serverExecuteInfo : serverExecuteInfoMap.keySet()) {
            String svc_name = serverExecuteInfoMap.get(serverExecuteInfo);
            String ser_id = svcSerMap.get(svc_name);
            serverExecuteInfo.setServer((ser_id != null) ? ser_id : "EMPTY");
            serverExecuteInfoList.add(serverExecuteInfo);
        }
        return serverExecuteInfoList.stream().sorted(Comparator.comparing(ServerExecuteInfo::getStd_deviation).reversed()).collect(Collectors.toList());
    }

    private void getSvcSvrMap() {
        String svc_names = svcNames.toString();
        if (svc_names.length()>0&&svc_names.substring(svc_names.length()-1,svc_names.length()).trim().equals(",")){
            svc_names = svc_names.substring(0,svc_names.length()-1);
        }
        String sql = "SELECT S.SUB_SYSTEM_ENAME || '-' || S.SRV_ID AS SRV_ID, C.SVC_NAME FROM TEA00 S JOIN TEA01 C ON S.SRV_ID = C.SRV_ID\n" +
            "WHERE C.SVC_NAME IN (%s) ";
        moduleNameService.getModule(partition, sql, svc_names, new ModuleNameCallBack() {

            @Override
            public void onSuccess(DataTable table) {
                svcNames = new StringBuilder("");
                DataRowCollection rows = table.getRows();
                rows.forEach(dataRow -> svcSerMap.put(dataRow.getValue("SVC_NAME").toString(), dataRow.getValue("SRV_ID").toString()));
            }

            @Override
            public void onFailure(String msg) {
                System.out.println(msg);
            }
        });
    }

}
